package com.chukun.flink.dataset.operator;

import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;

import java.util.ArrayList;
import java.util.List;

/**
 * @author chukun
 * @version 1.0.0
 * @description join 操作符的基本操作
 * @createTime 2022年05月28日 20:35:00
 */
public class JoinOperator {

    public static void main(String[] args) throws Exception {

        List<Tuple2<Integer, String>> commodityList = new ArrayList<>();
        List<Tuple2<String, Integer>> orderList = new ArrayList<>();

        commodityList.add(new Tuple2<>(1, "手机"));
        commodityList.add(new Tuple2<>(2, "电脑"));
        commodityList.add(new Tuple2<>(3, "移动电源"));

        orderList.add(new Tuple2<>("张三", 1));
        orderList.add(new Tuple2<>("赵六", 1));
        orderList.add(new Tuple2<>("李四", 2));
        orderList.add(new Tuple2<>("刘七", 3));
        orderList.add(new Tuple2<>("王五", 2));

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<Tuple2<Integer, String>> commodityDataSet = env.fromCollection(commodityList);
        DataSet<Tuple2<String, Integer>> orderDataSet = env.fromCollection(orderList);

        // 默认的连接方式
        DataSet<Tuple2<Tuple2<Integer, String>, Tuple2<String, Integer>>> defaultJoin =
                commodityDataSet.join(orderDataSet).where((key) -> key.f0).equalTo((key) -> key.f1);

        defaultJoin.print("defaultJoin");

        DataSet<Tuple2<String, String>> joinWithFunction =
                commodityDataSet.join(orderDataSet)
                        .where("f0")
                        .equalTo("f1")
                        // applying the JoinFunction on joining pairs
                        .with(new JoinFunction<Tuple2<Integer, String>, Tuple2<String, Integer>, Tuple2<String, String>>() {
                            @Override
                            public Tuple2<String, String> join(Tuple2<Integer, String> first, Tuple2<String, Integer> second) throws Exception {
                                return new Tuple2<>(first.f1, second.f0);
                            }
                        });

        joinWithFunction.print("joinWithFunction");

        env.execute("JoinOperator");
    }
}
